package org.apache.flink.runtime.webmonitor;

import java.io.File;
import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import javax.annotation.Nullable;

import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.allocation.ResourceAllocation;
import org.apache.flink.runtime.blob.TransientBlobService;
import org.apache.flink.runtime.leaderelection.LeaderContender;
import org.apache.flink.runtime.leaderelection.LeaderElectionService;
import org.apache.flink.runtime.resourceconsumer.ResourceConsumerGateway;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.rest.RestServerEndpoint;
import org.apache.flink.runtime.rest.handler.RestHandlerConfiguration;
import org.apache.flink.runtime.rest.handler.RestHandlerSpecification;
import org.apache.flink.runtime.rest.handler.cluster.ClusterOverviewHandler;
import org.apache.flink.runtime.rest.handler.cluster.ConsumerManagerCustomLogHandler;
import org.apache.flink.runtime.rest.handler.cluster.ConsumerManagerEnvironmentHandler;
import org.apache.flink.runtime.rest.handler.cluster.ConsumerManagerLogFileHandler;
import org.apache.flink.runtime.rest.handler.cluster.ConsumerManagerLogListHandler;
import org.apache.flink.runtime.rest.handler.cluster.ConsumerManagerThreadDumpHandler;
import org.apache.flink.runtime.rest.handler.cluster.ShutdownHandler;
import org.apache.flink.runtime.rest.handler.job.ConsumCancellationHandler;
import org.apache.flink.runtime.rest.handler.job.ConsumingStatusHandler;
import org.apache.flink.runtime.rest.handler.job.GeneratedLogUrlHandler;
import org.apache.flink.runtime.rest.handler.job.ConsumerManagerConsumerEnvironmentHandler;
import org.apache.flink.runtime.rest.handler.job.ResourceStatusHandler;
import org.apache.flink.runtime.rest.handler.job.metrics.ConsumerManagerMetricsHandler;
import org.apache.flink.runtime.rest.handler.job.metrics.ConsumerMetricsHandler;
import org.apache.flink.runtime.rest.handler.job.metrics.ProviderManagerMetricsHandler;
import org.apache.flink.runtime.rest.handler.taskmanager.ResourceProviderCustomLogHandler;
import org.apache.flink.runtime.rest.handler.taskmanager.ResourceProviderDetailsHandler;
import org.apache.flink.runtime.rest.handler.taskmanager.ResourceProviderLogFileHandler;
import org.apache.flink.runtime.rest.handler.taskmanager.ResourceProviderLogListHandler;
import org.apache.flink.runtime.rest.handler.taskmanager.ResourceProviderStdoutFileHandler;
import org.apache.flink.runtime.rest.handler.taskmanager.ResourceProviderThreadDumpHandler;
import org.apache.flink.runtime.rest.handler.taskmanager.ResourceProvidersHandler;
import org.apache.flink.runtime.rest.legacy.files.StaticFileServerHandler;
import org.apache.flink.runtime.rest.legacy.files.WebContentHandlerSpecification;
import org.apache.flink.runtime.rest.legacy.messages.ClusterOverviewHeaders;
import org.apache.flink.runtime.rest.legacy.metrics.MetricFetcher;
import org.apache.flink.runtime.rest.messages.ConsumingCancellationHeaders;
import org.apache.flink.runtime.rest.messages.ConsumerManagerEnvironmentHeaders;
import org.apache.flink.runtime.rest.messages.TerminationModeQueryParameter;
import org.apache.flink.runtime.rest.messages.cluster.ConsumerManagerCustomLogHeaders;
import org.apache.flink.runtime.rest.messages.cluster.ConsumerManagerLogFileHeader;
import org.apache.flink.runtime.rest.messages.cluster.ConsumerManagerLogListHeaders;
import org.apache.flink.runtime.rest.messages.cluster.ConsumerManagerStdoutFileHeader;
import org.apache.flink.runtime.rest.messages.cluster.ConsumerManagerThreadDumpHeaders;
import org.apache.flink.runtime.rest.messages.cluster.ShutdownHeaders;
import org.apache.flink.runtime.rest.messages.job.ConsumerManagerConsumerEnvironmentHeaders;
import org.apache.flink.runtime.rest.messages.job.ConsumingStatusInfoHeaders;
import org.apache.flink.runtime.rest.messages.job.ResourceStatusInfoHeaders;
import org.apache.flink.runtime.rest.messages.taskmanager.ResourceProvideCustomLogHeaders;
import org.apache.flink.runtime.rest.messages.taskmanager.ResourceProviderDetailsHeaders;
import org.apache.flink.runtime.rest.messages.taskmanager.ResourceProviderLogFileHeaders;
import org.apache.flink.runtime.rest.messages.taskmanager.ResourceProviderLogsHeaders;
import org.apache.flink.runtime.rest.messages.taskmanager.ResourceProviderStdoutFileHeaders;
import org.apache.flink.runtime.rest.messages.taskmanager.ResourceProviderThreadDumpHeaders;
import org.apache.flink.runtime.rest.messages.taskmanager.ProviderManagersHeaders;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.retriever.GatewayRetriever;
import org.apache.flink.runtime.webmonitor.history.ArchivedJson;
import org.apache.flink.runtime.webmonitor.history.JsonArchivist;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandler;
import org.apache.flink.util.ConfigurationException;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.flink.util.concurrent.FutureUtils;

/**
 * Rest endpoint which serves the web frontend REST calls.
 *
 * @param <T> type of the leader gateway
 */
/**
 * @类名: WebMonitorEndpoint
 * @说明: 
 *
 * @author   leehom
 * @Date	 2024年3月4日 上午11:35:31
 * 修改记录：
 *
 * @see 	 
 */
public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndpoint
        implements LeaderContender, JsonArchivist {

    protected final GatewayRetriever<? extends T> leaderRetriever;
    // 配置
    protected final Configuration clusterConfiguration;
    protected final RestHandlerConfiguration restConfiguration;
    // rpc网关
    private final GatewayRetriever<ResourceConsumerGateway> resourceConsumerRetriever;
    private final GatewayRetriever<ResourceManagerGateway> resourceManagerRetriever;
    // log存储，用于获取log的handler
    private final TransientBlobService transientBlobService;
    //
    protected final ScheduledExecutorService executor;

    private final MetricFetcher metricFetcher;

    private final LeaderElectionService leaderElectionService;

    private final FatalErrorHandler fatalErrorHandler;

    private boolean hasWebUI = false;

    private final Collection<JsonArchivist> archivingHandlers = new ArrayList<>(16);

    @Nullable private ScheduledFuture<?> executionGraphCleanupTask;

    public WebMonitorEndpoint(
            GatewayRetriever<? extends T> leaderRetriever,
            Configuration clusterConfiguration,
            RestHandlerConfiguration restConfiguration,
            GatewayRetriever<ResourceConsumerGateway> resourceConsumerRetriever,
            GatewayRetriever<ResourceManagerGateway> resourceManagerRetriever,
            TransientBlobService transientBlobService,
            ScheduledExecutorService executor,
            MetricFetcher metricFetcher,
            LeaderElectionService leaderElectionService,
            FatalErrorHandler fatalErrorHandler)
            throws IOException, ConfigurationException {
        super(clusterConfiguration);
        // 暂时屏蔽，以后去掉
        this.leaderRetriever = leaderRetriever; //Preconditions.checkNotNull(leaderRetriever);
        this.clusterConfiguration = Preconditions.checkNotNull(clusterConfiguration);
        this.restConfiguration = Preconditions.checkNotNull(restConfiguration);
        this.resourceConsumerRetriever = Preconditions.checkNotNull(resourceConsumerRetriever);
        this.resourceManagerRetriever = Preconditions.checkNotNull(resourceManagerRetriever);
        this.transientBlobService = Preconditions.checkNotNull(transientBlobService);
        this.executor = Preconditions.checkNotNull(executor);

        this.metricFetcher = metricFetcher;

        this.leaderElectionService = Preconditions.checkNotNull(leaderElectionService);
        this.fatalErrorHandler = Preconditions.checkNotNull(fatalErrorHandler);
    }

    @Override
    protected List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> initializeHandlers(
            final CompletableFuture<String> localAddressFuture) {
        ArrayList<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers =
                new ArrayList<>(30);

        final Collection<Tuple2<RestHandlerSpecification, ChannelInboundHandler>>
                webSubmissionHandlers = initializeWebSubmissionHandlers(localAddressFuture);
        handlers.addAll(webSubmissionHandlers);
        final boolean hasWebSubmissionHandlers = !webSubmissionHandlers.isEmpty();

        final Duration asyncOperationStoreDuration =
                clusterConfiguration.get(RestOptions.ASYNC_OPERATION_STORE_DURATION);
        final Time timeout = restConfiguration.getTimeout();
        // ******状态
        // 资源统计, 对应ResourceAllocationStatistics
        ClusterOverviewHandler clusterOverviewHandler =
                new ClusterOverviewHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ClusterOverviewHeaders.getInstance(),
                        resourceManagerRetriever);
        // 消费状态
        ConsumingStatusHandler consumingStatusHandler =
                new ConsumingStatusHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ConsumingStatusInfoHeaders.getInstance(),
                        resourceConsumerRetriever);
        
        // 资源状态 ResourceStatus
        ResourceStatusHandler resourceStatusHandler =
                new ResourceStatusHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ResourceStatusInfoHeaders.getInstance(),
                        resourceManagerRetriever
                        );


        // ******环境信息
        ConsumerManagerEnvironmentHandler consumerManagerEnvironmentHandler =
                new ConsumerManagerEnvironmentHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ConsumerManagerEnvironmentHeaders.getInstance());

        ConsumerManagerConsumerEnvironmentHandler consumerManagerConsumerEnvironmentHandler =
                new ConsumerManagerConsumerEnvironmentHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ConsumerManagerConsumerEnvironmentHeaders.getInstance());

        // 资源提供者
        ResourceProvidersHandler resourceProvidersHandler =
                new ResourceProvidersHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ProviderManagersHeaders.getInstance(),
                        resourceManagerRetriever);

        ResourceProviderDetailsHandler resourceProviderDetailsHandler =
                new ResourceProviderDetailsHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ResourceProviderDetailsHeaders.getInstance(),
                        resourceManagerRetriever);

        // *******metircs
        // 去掉，性能差，使用reporter输出
//        final ConsumerManagerMetricsHandler consumerManagerMetricsHandler =
//                new ConsumerManagerMetricsHandler(
//                        leaderRetriever, timeout, responseHeaders, metricFetcher);
//        
//        final ConsumerMetricsHandler consumerMetricsHandler =
//                new ConsumerMetricsHandler(leaderRetriever, timeout, responseHeaders, metricFetcher);
//
//        final ProviderManagerMetricsHandler providerManagerMetricsHandler =
//                new ProviderManagerMetricsHandler(
//                        leaderRetriever, timeout, responseHeaders, metricFetcher);
        
        // ******操作类
        final ConsumCancellationHandler cancelTerminationHandler =
                new ConsumCancellationHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ConsumingCancellationHeaders.getInstance(),
                        TerminationModeQueryParameter.TerminationMode.CANCEL);

        // this is kept just for legacy reasons. STOP has been replaced by STOP-WITH-SAVEPOINT.
        final ConsumCancellationHandler stopTerminationHandler =
                new ConsumCancellationHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ConsumingCancellationHeaders.getInstance(),
                        TerminationModeQueryParameter.TerminationMode.STOP);

        // 
//        final GeneratedLogUrlHandler jobManagerLogUrlHandler =
//                new GeneratedLogUrlHandler(
//                        localAddressFuture.thenApply(url -> url + "/#/job-manager/logs"));

        final ShutdownHandler shutdownHandler =
                new ShutdownHandler(
                        leaderRetriever, timeout, responseHeaders, ShutdownHeaders.getInstance());

        final File webUiDir = restConfiguration.getWebUiDir();

        Optional<StaticFileServerHandler<T>> optWebContent;

        try {
            optWebContent = WebMonitorUtils.tryLoadWebContent(leaderRetriever, timeout, webUiDir);
        } catch (IOException e) {
            log.warn("Could not load web content handler.", e);
            optWebContent = Optional.empty();
        }

        handlers.add(Tuple2.of(clusterOverviewHandler.getMessageHeaders(), clusterOverviewHandler));
        handlers.add(
                Tuple2.of(
                        consumerManagerEnvironmentHandler.getMessageHeaders(),
                        consumerManagerEnvironmentHandler));
        handlers.add(
                Tuple2.of(
                        consumerManagerConsumerEnvironmentHandler.getMessageHeaders(),
                        consumerManagerConsumerEnvironmentHandler));
        handlers.add(Tuple2.of(consumingStatusHandler.getMessageHeaders(), consumingStatusHandler));
        handlers.add(Tuple2.of(resourceStatusHandler.getMessageHeaders(), resourceStatusHandler));

        handlers.add(Tuple2.of(resourceProvidersHandler.getMessageHeaders(), resourceProvidersHandler));
        handlers.add(
                Tuple2.of(
                        resourceProviderDetailsHandler.getMessageHeaders(), resourceProviderDetailsHandler));
        
//        handlers.add(Tuple2.of(consumerMetricsHandler.getMessageHeaders(), consumerMetricsHandler));
//        handlers.add(
//                Tuple2.of(
//                        providerManagerMetricsHandler.getMessageHeaders(), providerManagerMetricsHandler));
//        handlers.add(
//                Tuple2.of(consumerManagerMetricsHandler.getMessageHeaders(), consumerManagerMetricsHandler));

        handlers.add(
                Tuple2.of(
                        cancelTerminationHandler.getMessageHeaders(),
                        cancelTerminationHandler));
        handlers.add(Tuple2.of(shutdownHandler.getMessageHeaders(), shutdownHandler));

        optWebContent.ifPresent(
                webContent -> {
                    handlers.add(
                            Tuple2.of(WebContentHandlerSpecification.getInstance(), webContent));
                    hasWebUI = true;
                });

        // ******日志文件类
        // load the log and stdout file handler for the main cluster component
        final WebMonitorUtils.LogFileLocation logFileLocation =
                WebMonitorUtils.LogFileLocation.find(clusterConfiguration);

        final ConsumerManagerLogFileHandler consumerManagerLogFileHandler =
                new ConsumerManagerLogFileHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ConsumerManagerLogFileHeader.getInstance(),
                        logFileLocation.logFile);

        final ConsumerManagerLogFileHandler consumerManagerStdoutFileHandler =
                new ConsumerManagerLogFileHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ConsumerManagerStdoutFileHeader.getInstance(),
                        logFileLocation.stdOutFile);

        final ConsumerManagerCustomLogHandler consumerManagerCustomLogHandler =
                new ConsumerManagerCustomLogHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ConsumerManagerCustomLogHeaders.getInstance(),
                        logFileLocation.logDir);

        final ConsumerManagerLogListHandler consumerManagerLogListHandler =
                new ConsumerManagerLogListHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ConsumerManagerLogListHeaders.getInstance(),
                        logFileLocation.logDir);

        // 线程dump
        final ConsumerManagerThreadDumpHandler consumerManagerThreadDumpHandler =
                new ConsumerManagerThreadDumpHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ConsumerManagerThreadDumpHeaders.getInstance(),
                        resourceConsumerRetriever
                        );

        handlers.add(Tuple2.of(ConsumerManagerLogFileHeader.getInstance(), consumerManagerLogFileHandler));
        handlers.add(
                Tuple2.of(ConsumerManagerStdoutFileHeader.getInstance(), consumerManagerStdoutFileHandler));
        handlers.add(
                Tuple2.of(ConsumerManagerCustomLogHeaders.getInstance(), consumerManagerCustomLogHandler));
        handlers.add(Tuple2.of(ConsumerManagerLogListHeaders.getInstance(), consumerManagerLogListHandler));
        handlers.add(
                Tuple2.of(ConsumerManagerThreadDumpHeaders.getInstance(), consumerManagerThreadDumpHandler));

        // Resource Provider log and stdout file handler
        final Time cacheEntryDuration = Time.milliseconds(restConfiguration.getRefreshInterval());

        final ResourceProviderLogFileHandler resourceProviderLogFileHandler =
                new ResourceProviderLogFileHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ResourceProviderLogFileHeaders.getInstance(),
                        resourceManagerRetriever,
                        transientBlobService,
                        cacheEntryDuration);

        final ResourceProviderStdoutFileHandler resourceProviderStdoutFileHandler =
                new ResourceProviderStdoutFileHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ResourceProviderStdoutFileHeaders.getInstance(),
                        resourceManagerRetriever,
                        transientBlobService,
                        cacheEntryDuration);

        final ResourceProviderCustomLogHandler resourceProviderCustomLogHandler =
                new ResourceProviderCustomLogHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ResourceProvideCustomLogHeaders.getInstance(),
                        resourceManagerRetriever,
                        transientBlobService,
                        cacheEntryDuration);

        final ResourceProviderLogListHandler resourceProviderLogListHandler =
                new ResourceProviderLogListHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ResourceProviderLogsHeaders.getInstance(),
                        resourceManagerRetriever);

        final ResourceProviderThreadDumpHandler resourceProviderThreadDumpFileHandler =
                new ResourceProviderThreadDumpHandler(
                        leaderRetriever,
                        timeout,
                        responseHeaders,
                        ResourceProviderThreadDumpHeaders.getInstance(),
                        resourceManagerRetriever);

        handlers.add(Tuple2.of(ResourceProviderLogFileHeaders.getInstance(), resourceProviderLogFileHandler));
        handlers.add(
                Tuple2.of(
                        ResourceProviderStdoutFileHeaders.getInstance(), resourceProviderStdoutFileHandler));
        handlers.add(
                Tuple2.of(ResourceProvideCustomLogHeaders.getInstance(), resourceProviderCustomLogHandler));
        handlers.add(Tuple2.of(ResourceProviderLogsHeaders.getInstance(), resourceProviderLogListHandler));
        handlers.add(
                Tuple2.of(
                        ResourceProviderThreadDumpHeaders.getInstance(),
                        resourceProviderThreadDumpFileHandler));

        handlers.stream()
                .map(tuple -> tuple.f1)
                .filter(handler -> handler instanceof JsonArchivist)
                .forEachOrdered(handler -> archivingHandlers.add((JsonArchivist) handler));

        return handlers;
    }

    protected Collection<Tuple2<RestHandlerSpecification, ChannelInboundHandler>>
            initializeWebSubmissionHandlers(final CompletableFuture<String> localAddressFuture) {
        return Collections.emptyList();
    }

    @Override
    public void startInternal() throws Exception {
    	// 启动领导选举服务
        leaderElectionService.start(this);
        if (hasWebUI) {
            log.info("Web frontend listening at {}.", getRestBaseUrl());
        }
    }

    @Override
    protected CompletableFuture<Void> shutDownInternal() {
        if (executionGraphCleanupTask != null) {
            executionGraphCleanupTask.cancel(false);
        }

        final CompletableFuture<Void> shutdownFuture =
                FutureUtils.runAfterwards(
                        super.shutDownInternal(),
                        () -> ExecutorUtils.gracefulShutdown(10, TimeUnit.SECONDS, executor));

        final File webUiDir = restConfiguration.getWebUiDir();

        return FutureUtils.runAfterwardsAsync(
                shutdownFuture,
                () -> {
                    Exception exception = null;
                    try {
                        log.info("Removing cache directory {}", webUiDir);
                        FileUtils.deleteDirectory(webUiDir);
                    } catch (Exception e) {
                        exception = e;
                    }

                    try {
                        leaderElectionService.stop();
                    } catch (Exception e) {
                        exception = ExceptionUtils.firstOrSuppressed(e, exception);
                    }

                    if (exception != null) {
                        throw exception;
                    }
                });
    }

    // -------------------------------------------------------------------------
    // LeaderContender
    // -------------------------------------------------------------------------
    @Override
    public void grantLeadership(final UUID leaderSessionID) {
        log.info(
                "{} was granted leadership with leaderSessionID={}",
                getRestBaseUrl(),
                leaderSessionID);
        leaderElectionService.confirmLeadership(leaderSessionID, getRestBaseUrl());
    }

    @Override
    public void revokeLeadership() {
        log.info("{} lost leadership", getRestBaseUrl());
    }

    @Override
    public String getDescription() {
        return getRestBaseUrl();
    }

    @Override
    public void handleError(final Exception exception) {
        fatalErrorHandler.onFatalError(exception);
    }

    @Override
    public Collection<ArchivedJson> archiveJsonWithPath(ResourceAllocation resourceAllocation)
            throws IOException {
        Collection<ArchivedJson> archivedJson = new ArrayList<>(archivingHandlers.size());
        for (JsonArchivist archivist : archivingHandlers) {
            Collection<ArchivedJson> subArchive = archivist.archiveJsonWithPath(resourceAllocation);
            archivedJson.addAll(subArchive);
        }
        return archivedJson;
    }

    public static ScheduledExecutorService createExecutorService(
            int numThreads, int threadPriority, String componentName) {
        if (threadPriority < Thread.MIN_PRIORITY || threadPriority > Thread.MAX_PRIORITY) {
            throw new IllegalArgumentException(
                    String.format(
                            "The thread priority must be within (%s, %s) but it was %s.",
                            Thread.MIN_PRIORITY, Thread.MAX_PRIORITY, threadPriority));
        }

        return Executors.newScheduledThreadPool(
                numThreads,
                new ExecutorThreadFactory.Builder()
                        .setThreadPriority(threadPriority)
                        .setPoolName("Flink-" + componentName)
                        .build());
    }
}
